package cn.tukk.sparkproject.spark.product;

import cn.tukk.sparkproject.conf.ConfigurationManager;
import cn.tukk.sparkproject.constant.Constants;
import cn.tukk.sparkproject.dao.IAreaTop3ProductDAO;
import cn.tukk.sparkproject.dao.ITaskDAO;
import cn.tukk.sparkproject.dao.factory.DAOFactory;
import cn.tukk.sparkproject.domain.AreaTop3Product;
import cn.tukk.sparkproject.domain.Task;
import cn.tukk.sparkproject.spark.product.MyUDF.MyGroupConcatDistinctUDAF;
import cn.tukk.sparkproject.util.ParamUtils;
import cn.tukk.sparkproject.util.SparkUtils;
import com.alibaba.fastjson.JSONObject;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 各区域top3热门商品统计Spark作业
 * @author Administrator
 *
 */
public class MyAreaTop3ProductSpark {
    public static void main(String[] args) {
        //第一步设置sparkconf
        SparkConf sparkConf = new SparkConf().setAppName("AreaTop3Product");
        SparkUtils.setMaster(sparkConf);//如果配置local为true本地模式，否则使用集群模式
        //生成sqlcontext  local模式或者hivecontext  生产模式
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        SQLContext sqlContext = SparkUtils.getSQLContext(jsc.sc());

        //注册自定义函数
        // 注册自定义函数
        sqlContext.udf().register("concat_long_string",
                new ConcatLongStringUDF(), DataTypes.StringType);
        sqlContext.udf().register("get_json_object",
                new GetJsonObjectUDF(), DataTypes.StringType);
        sqlContext.udf().register("random_prefix",
                new RandomPrefixUDF(), DataTypes.StringType);
        sqlContext.udf().register("remove_random_prefix",
                new RemoveRandomPrefixUDF(), DataTypes.StringType);
        sqlContext.udf().register("group_concat_distinct",
                new GroupConcatDistinctUDAF());


        //模拟数据生成
        SparkUtils.mockData(jsc,sqlContext);
        //根据taskid查询  筛选的task参数
        ITaskDAO taskDAO = DAOFactory.getTaskDAO();
        long taskid = ParamUtils.getTaskIdFromArgs(args, Constants.SPARK_LOCAL_TASKID_PRODUCT);
        Task task = taskDAO.findById(taskid);
        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());
        String startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE);
        String endDate = ParamUtils.getParam(taskParam,Constants.PARAM_END_DATE);

        // 技术点1   hive源根据指定的查询条件  通过hivecontext查询  hive表  查询用户点击行为
        JavaPairRDD<Long,Row>   cityid2clickActionRDD=getcityid2ClickActionRDDByDate(sqlContext,startDate,endDate);
        System.out.println("cityid2clickActionRDD:"+cityid2clickActionRDD.count());
        //技术点2   从mysql查询数据  以后数据源
        JavaPairRDD<Long,Row> cityid2cityInfoRDD=getcityid2CityInfoRDD(sqlContext);
        System.out.println("cityid2cityInfoRDD"+ cityid2cityInfoRDD.count());
        //技术点3   将rdd转化为dataframe 并注册临时表    生成商品基础信息临时表
        generateTempClickProductBasicTable(sqlContext,cityid2clickActionRDD,cityid2cityInfoRDD);

        //生成各区域各商品点击次数的临时表
        generateTempAreaProductClickCountTable( sqlContext);
        
        //生成包含完整商品信息的各区域商品点击次数临时表
        generateTempAreaFullProductClickCountTable(sqlContext);

        //使用开窗函数获取各个区域点击次数排名前三的热门商品
        JavaRDD<Row> areaTop3ProductRDD=getAreaTop3ProductRDD(sqlContext);
        System.out.println("areaTop3ProductRDD: " + areaTop3ProductRDD.count());

        // 这边的写入mysql和之前不太一样
        // 因为实际上，就这个业务需求而言，计算出来的最终数据量是比较小的
        // 总共就不到10个区域，每个区域还是top3热门商品，总共最后数据量也就是几十个
        // 所以可以直接将数据collect()到本地
        // 用批量插入的方式，一次性插入mysql即可
        List<Row> rows = areaTop3ProductRDD.collect();
        System.out.println("rows: " + rows.size());
        persistAreaTop3Product(taskid, rows);

        jsc.close();


    }

    private static void persistAreaTop3Product(long taskid, List<Row> rows) {
        List<AreaTop3Product> areaTop3Products = new ArrayList<AreaTop3Product>();

        for(Row row : rows) {
            AreaTop3Product areaTop3Product = new AreaTop3Product();
            areaTop3Product.setTaskid(taskid);
            areaTop3Product.setArea(row.getString(0));
            areaTop3Product.setAreaLevel(row.getString(1));
            areaTop3Product.setProductid(row.getLong(2));
            areaTop3Product.setClickCount(Long.valueOf(String.valueOf(row.get(3))));
            areaTop3Product.setCityInfos(row.getString(4));
            areaTop3Product.setProductName(row.getString(5));
            areaTop3Product.setProductStatus(row.getString(6));
            areaTop3Products.add(areaTop3Product);
        }

        IAreaTop3ProductDAO areTop3ProductDAO = DAOFactory.getAreaTop3ProductDAO();
        areTop3ProductDAO.insertBatch(areaTop3Products);
    }

    private static JavaRDD<Row> getAreaTop3ProductRDD(SQLContext sqlContext) {
        // 技术点：开窗函数

        // 使用开窗函数先进行一个子查询
        // 按照area进行分组，给每个分组内的数据，按照点击次数降序排序，打上一个组内的行号
        // 接着在外层查询中，过滤出各个组内的行号排名前3的数据
        // 其实就是咱们的各个区域下top3热门商品

        // 华北、华东、华南、华中、西北、西南、东北
        // A级：华北、华东
        // B级：华南、华中
        // C级：西北、西南
        // D级：东北

        // case when
        // 根据多个条件，不同的条件对应不同的值
        // case when then ... when then ... else ... end
        String sql =
                "SELECT "
                        + "area,"
                        + "CASE "
                        + "WHEN area='China North' OR area='China East' THEN 'A Level' "
                        + "WHEN area='China South' OR area='China Middle' THEN 'B Level' "
                        + "WHEN area='West North' OR area='West South' THEN 'C Level' "
                        + "ELSE 'D Level' "
                        + "END area_level,"
                        + "product_id,"
                        + "click_count,"
                        + "city_infos,"
                        + "product_name,"
                        + "product_status "
                        + "FROM ("
                        + "SELECT "
                        + "area,"
                        + "product_id,"
                        + "click_count,"
                        + "city_infos,"
                        + "product_name,"
                        + "product_status,"
                        + "row_number() OVER (PARTITION BY area ORDER BY click_count DESC) rank "
                        + "FROM tmp_area_fullprod_click_count "
                        + ") t "
                        + "WHERE rank<=3";

        DataFrame df = sqlContext.sql(sql);

        return df.javaRDD();



    }

    /**
     * 生成商品点击次数临时表
     */
    private static void generateTempAreaFullProductClickCountTable(SQLContext sqlContext) {
        // 将之前得到的各区域各商品点击次数表，product_id
        // 去关联商品信息表，product_id，product_name和product_status
        // product_status要特殊处理，0，1，分别代表了自营和第三方的商品，放在了一个json串里面
        // get_json_object()函数，可以从json串中获取指定的字段的值
        // if()函数，判断，如果product_status是0，那么就是自营商品；如果是1，那么就是第三方商品
        // area, product_id, click_count, city_infos, product_name, product_status

        // 为什么要费时费力，计算出来商品经营类型
        // 你拿到到了某个区域top3热门的商品，那么其实这个商品是自营的，还是第三方的
        // 其实是很重要的一件事

        // 技术点：内置if函数的使用
        String sql =
                "SELECT "
                        + "tapcc.area,"
                        + "tapcc.product_id,"
                        + "tapcc.click_count,"
                        + "tapcc.city_infos,"
                        + "pi.product_name,"
                        + "if(get_json_object(pi.extend_info,'product_status')='0','Self','Third Party') product_status "
                        + "FROM tmp_area_product_click_count tapcc "
                        + "JOIN product_info pi ON tapcc.product_id=pi.product_id ";
        DataFrame df = sqlContext.sql(sql);
        System.out.println("tmp_area_fullprod_click_count: " + df.count());
        df.registerTempTable("tmp_area_fullprod_click_count");

    }

    /**
     * 生成区域商品点击次数临时表 包含商品完整信息
     */


    /**
     * 生成各区域商品点击次数临时表
     * @param sqlContext
     */

    private static void generateTempAreaProductClickCountTable(SQLContext sqlContext) {
        // 按照area和product_id两个字段进行分组
        // 计算出各区域各商品的点击次数
        // 可以获取到每个area下的每个product_id的城市信息拼接起来的串
        String sql =
                "SELECT "
                        + "area,"
                        + "product_id,"
                        + "count(*) click_count, "
                        + "group_concat_distinct(concat_long_string(city_id,city_name,':')) city_infos "
                        + "FROM tmp_click_product_basic "
                        + "GROUP BY area,product_id ";
        DataFrame df = sqlContext.sql(sql);
        System.out.println("tmp_area_product_click_count:"+df.count());
        //再次将查询出来的数据注册为一个临时表   各区域各商品的点击次数
        df.registerTempTable("temp_area_product_click_count");

    }


    /**
     *
     * @param sqlContext
     * @param cityid2clickActionRDD
     * @param cityid2cityInfoRDD
     */
    private static void generateTempClickProductBasicTable(SQLContext sqlContext, JavaPairRDD<Long,Row> cityid2clickActionRDD, JavaPairRDD<Long,Row> cityid2cityInfoRDD) {
        //执行join操作，进行点击行为数据和城市数据关联
        JavaPairRDD<Long, Tuple2<Row, Row>> joinedRDD = cityid2clickActionRDD.join(cityid2cityInfoRDD);
        //将join之后的rdd转换为javaRDd<row>这样才能转化为DataFrame
        JavaRDD<Row> mappedRDD = joinedRDD.map(tuple ->  //自己尝试换位lambda表达式吧
//            long cityid = tuple._1;
//            Row clickAction = tuple._2._1;
//            Row cityInfo = tuple._2._2;
//
//            long productid = clickAction.getLong(1);
//            String cityName = cityInfo.getString(1);
//            String area = cityInfo.getString(2);
            RowFactory.create(tuple._1,tuple._2._2.getString(1), tuple._2._2.getString(2),tuple._2._1.getLong(1)));

        //基于JavaRDD<ROW>的格式，就可以将其转化为DataFrame
        ArrayList<StructField> structFields = new ArrayList<StructField>();
        structFields.add(DataTypes.createStructField("city_id",DataTypes.LongType,true));
        structFields.add(DataTypes.createStructField("city_name",DataTypes.StringType,true));
        structFields.add(DataTypes.createStructField("area",DataTypes.StringType,true));
        structFields.add(DataTypes.createStructField("product_id",DataTypes.LongType,true));

        //0北京
        //1上海
        //0 北京
        //group by area,product_id
        //0:北京，1：上海
        //两个函数
        //UDF concat2() 将两个字段拼接起来，用指定分隔符
        //UDAF group_concat_distinct() 将一个分组中的多个字段值，用逗号拼接起来，同时去重

        StructType schema = DataTypes.createStructType(structFields);
        DataFrame df = sqlContext.createDataFrame(mappedRDD, schema);
        System.out.println("tmp_click_product_basic:"+df.count());
        //将DataFrame数据，注册成临时表 （tmp_click_product_basic)
        df.registerTempTable("tmp_click_product_basic");

    }

    /**
     * 城市信息  从mysql获取数据
     * @param sqlContext
     * @return
     */
    private static JavaPairRDD<Long,Row> getcityid2CityInfoRDD(SQLContext sqlContext) {
        //参考官网从mysql获取配置信息
        String url;
        String user;
        String password;
        boolean local=ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
        if(local){
            url=ConfigurationManager.getProperty(Constants.JDBC_URL);
            user=ConfigurationManager.getProperty(Constants.JDBC_USER);
            password=ConfigurationManager.getProperty(Constants.JDBC_PASSWORD);
        }else{
            url = ConfigurationManager.getProperty(Constants.JDBC_URL_PROD);
            user = ConfigurationManager.getProperty(Constants.JDBC_USER_PROD);
            password = ConfigurationManager.getProperty(Constants.JDBC_PASSWORD_PROD);
        }
        Map<String, String> options = new HashMap<>();
        options.put("url",url);
        options.put("dbtable","city_info");
        options.put("user",user);
        options.put("password",password);

        //通过sqlContext从mysql中查询数据
        DataFrame cityInfoDF = sqlContext.read().format("jdbc").options(options).load();
        JavaRDD<Row> cityInfoRowRDD = cityInfoDF.javaRDD();
        return cityInfoRowRDD.mapToPair(row -> new Tuple2<Long, Row>(Long.valueOf(String.valueOf(row.get(0))), row));
    }

    /**
     * 查询用户行为rdd
     * @param sqlContext
     * @param startDate
     * @param endDate
     * @return
     */
    private static JavaPairRDD<Long,Row> getcityid2ClickActionRDDByDate(SQLContext sqlContext, String startDate, String endDate) {
        //通过sql 最直接从hive表中查询user_visit_action中，用户行为数据
        //第一个限定：click_product_id,限定不为空的访问行为，
        //第二个限定，在用户指定日期范围内数据
        String sql="SELECT "
                + "city_id,"
                + "click_product_id product_id "
                + "FROM user_visit_action "
                + "WHERE click_product_id IS NOT NULL "
                + "AND date>='" + startDate + "' "
                + "AND date<='" + endDate + "'";
        DataFrame clickActionDF = sqlContext.sql(sql);//DataFrame
        JavaRDD<Row> clickActionRow = clickActionDF.javaRDD();
//       JavaPairRDD<Long, Row> cityid2clickActionRDD = clickActionRow.mapToPair(new PairFunction<Row, Long, Row>() {
//            private static final long serialVersionUID = -1674204026801424299L;
//            @Override
//            public Tuple2<Long, Row> call(Row row) throws Exception {
//                Long cityid = row.getLong(0);
//                return new Tuple2<Long,Row>(cityid,row);
//            }
//        });
        return clickActionRow.mapToPair(row ->
                new Tuple2<>(row.getLong(0), row)//row.getLong(0) 代表city
        );
    }

}
